package com.mxp.influx.client.service.impl;

import com.mxp.influx.client.entity.ClientConnection;
import com.mxp.influx.client.entity.InfluxDBInfo;
import com.mxp.influx.client.service.ClientConnectionService;
import com.mxp.influx.client.service.InfluxDBService;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.BatchOptions;
import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
import org.influxdb.dto.Pong;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.Executors;

/**
 * @author maoxp
 * @date 2024-12-27 16:49
 */
@Slf4j
@Service
public class InfluxDBServiceImpl implements InfluxDBService {

    /**
     * 所有连接集合
     */
    @Resource
    private Map<String, InfluxDBInfo> influxDBConnections;

    /**
     * 连接服务
     */
    @Resource
    private ClientConnectionService clientConnectionService;

    /**
     * 获取InfluxDB
     * @param connectionId
     * @return
     */
    @Override
    public InfluxDB getInfluxDB(String connectionId) {
        InfluxDBInfo influxDBInfo = influxDBConnections.get(connectionId);
        if (influxDBInfo == null) {
            ClientConnection clientConnection = clientConnectionService.getById(connectionId);
            influxDBInfo = new InfluxDBInfo();
            influxDBInfo.setInfluxDB(this.connectionInfluxDB(clientConnection));
            influxDBConnections.put(connectionId, influxDBInfo);
        }
        influxDBInfo.setCreateDateTime(new Date().getTime());
        return influxDBInfo.getInfluxDB();
    }

    /**
     * 创建InfluxDB对象
     * @param clientConnection
     * @return
     */
    private InfluxDB createInfluxDB(ClientConnection clientConnection) {
        // 连接到 influxDB
        String protocolType = "http://";
        if (clientConnection.getUseSsl() != null && clientConnection.getUseSsl() == 1) {
            protocolType = "https://";
        }
        InfluxDB influxDB = InfluxDBFactory.connect(
                protocolType + clientConnection.getIp() + ":" + clientConnection.getPort(),
                clientConnection.getUsername(),
                clientConnection.getPassword());
        return influxDB;
    }

    /**
     * 创建并连接InfluxDB
     * @param clientConnection
     * @return
     */
    @Override
    public InfluxDB connectionInfluxDB(ClientConnection clientConnection) {
        // 创建InfluxDB对象
        InfluxDB influxDB = this.createInfluxDB(clientConnection);
        // 设置日志级别
        influxDB.setLogLevel(InfluxDB.LogLevel.BASIC);
        // 配置批量写入
        influxDB.enableBatch(BatchOptions.DEFAULTS
                .actions(100)
                .flushDuration(100)
                .bufferLimit(10)
                .exceptionHandler((points, e) -> {
                    StringBuilder msgBuilder = new StringBuilder("写分失败: ");
                    for (Point point : points) {
                        msgBuilder.append(point.toString()).append(", ");
                    }
                    String msg = msgBuilder.toString();
                    if (msg.length() > 10000) {
                        msg = msg.substring(0, 10000);
                    }
                    log.error(msg, e);
                })
                .threadFactory(Executors.defaultThreadFactory()));
        // 测试
        influxDB.ping();
        return influxDB;
    }

    /**
     * 测试连接
     * @param clientConnection
     * @return
     */
    @Override
    public String testInfluxDB(ClientConnection clientConnection) {
        InfluxDB influxDB = this.createInfluxDB(clientConnection);
        Pong ping = influxDB.ping();
        if (ping == null) {
            return null;
        }
        String version = ping.getVersion();
        influxDB.close();
        return version;
    }

    /**
     * 断开连接
     * @param connectionId
     */
    @Override
    public void disconnectInfluxDB(String connectionId) {
        InfluxDBInfo influxDBInfo = influxDBConnections.get(connectionId);
        if (influxDBInfo != null) {
            InfluxDB influxDB = influxDBInfo.getInfluxDB();
            influxDB.close();
            influxDBConnections.remove(connectionId);
        }
    }

    /**
     * 断开超时连接
     * @param duration
     */
    @Override
    public void disconnectOvertimeInfluxDB(Long duration) {
        List<String> keys = new ArrayList<>(influxDBConnections.keySet());
        long nowTime = new Date().getTime();
        for (String key : keys) {
            InfluxDBInfo influxDBInfo = influxDBConnections.get(key);
            // 一小时不使用就关闭连接
            if (nowTime - influxDBInfo.getCreateDateTime() >= duration) {
                this.disconnectInfluxDB(key);
            }
        }
    }
}
